Qwen3-VL-2B 多模态支持

需求概述

在 mini-sglang 中支持 Qwen/Qwen3-VL-2B-Instruct 模型的图片理解能力。第一阶段约束：仅支持图片(不支持视频)、仅 TP=1、仅 eager 模式(多模态 batch 禁用 cuda graph)、仅 /v1/chat/completions 图片输入。

技术架构

整体策略是"复用现有 Qwen3 文本主干 + 新增最小多模态补丁层"：

HTTP 请求 (OpenAI-style content with image_url)
  │
  ▼
API Server (解析多段 content，提取图片 URL/base64)
  │
  ▼
Tokenizer Worker (AutoProcessor: chat_template + pixel_values + image_grid_thw + input_ids)
  │  同时计算 MRoPE positions 和 position_delta
  ▼
Scheduler (携带多模态字段创建 Req/Batch，对多模态 batch 禁用 cuda graph)
  │
  ▼
Model Forward:
  ├─ Vision Encoder (PatchEmbed → VisionBlocks → PatchMerger + DeepStack)
  ├─ Embedding 替换 (image placeholder → vision embeddings)
  ├─ DeepStack 注入 (在指定 decoder layer 注入 deepstack embeddings)
  ├─ MRoPE (3D positions → 按 mrope_section 切分 cos/sin)
  └─ Text Decoder (复用现有 Qwen3 decoder layers)

子需求分解

按照计划文档的阶段划分，共 7 个子需求：

子需求 1：配置、注册、依赖补齐

场景：让 mini-sglang 能识别 Qwen3-VL-2B 配置，为后续模块提供字段。

涉及文件：

文件	修改类型	影响函数/区域
`python/minisgl/models/register.py`	修改	`_MODEL_REGISTRY` 字典
`python/minisgl/models/config.py`	修改	`ModelConfig` 类、`from_hf()` 方法
`python/minisgl/utils/hf.py`	修改	新增 `load_processor()`
`pyproject.toml`	修改	dependencies
`python/minisgl/multimodal/__init__.py`	新增	包初始化
`python/minisgl/multimodal/schema.py`	新增	数据结构定义

实现细节：

模型注册 (register.py:5)

在 _MODEL_REGISTRY 新增：

"Qwen3VLForConditionalGeneration": (".qwen3_vl", "Qwen3VLForConditionalGeneration"),

配置扩展 (config.py)

ModelConfig 需要新增字段来承载 VL 专属配置。由于当前 ModelConfig 是 frozen=True 的 dataclass，需要新增以下字段(带默认值)：

@dataclass(frozen=True)
class ModelConfig:
    # ... 现有字段 ...
    image_token_id: int | None = None
    video_token_id: int | None = None
    vision_start_token_id: int | None = None
    vision_end_token_id: int | None = None
    mrope_section: list[int] | None = None
    vision_config: dict | None = None  # 存储原始 vision_config 字典
    is_multimodal: bool = False

from_hf() 需要在展开 text_config 之前，先从 top 提取多模态字段：

@classmethod
def from_hf(cls, config: PretrainedConfig) -> ModelConfig:
    # 先提取多模态字段
    image_token_id = getattr(config, "image_token_id", None)
    video_token_id = getattr(config, "video_token_id", None)
    vision_start_token_id = getattr(config, "vision_start_token_id", None)
    vision_end_token_id = getattr(config, "vision_end_token_id", None)
    vision_config_raw = None
    if hasattr(config, "vision_config") and config.vision_config is not None:
        vc = config.vision_config
        vision_config_raw = vc.to_dict() if hasattr(vc, "to_dict") else dict(vc)

    # 从 rope_scaling 中提取 mrope_section
    top_rope_scaling = getattr(config, "rope_scaling", None)
    if top_rope_scaling is None and hasattr(config, "text_config"):
        top_rope_scaling = getattr(config.text_config, "rope_scaling", None)
    mrope_section = None
    if top_rope_scaling and "mrope_section" in top_rope_scaling:
        mrope_section = top_rope_scaling["mrope_section"]

    is_multimodal = image_token_id is not None

    # 然后展开 text_config (现有逻辑)
    if hasattr(config, "text_config") and config.text_config is not None:
        top = config
        config = config.text_config
        for attr in ("architectures", "rope_theta", "rope_scaling"):
            if not getattr(config, attr, None) and getattr(top, attr, None):
                setattr(config, attr, getattr(top, attr))
    # ... 构造 ModelConfig，包含新字段 ...

Processor 加载 (utils/hf.py)

def load_processor(model_path: str):
    from transformers import AutoProcessor
    return AutoProcessor.from_pretrained(model_path)

依赖 (pyproject.toml)

新增 Pillow 和 requests (如未包含) 到 dependencies。einops 也需要添加(视觉编码器用到)。

多模态数据结构 (multimodal/schema.py)

@dataclass
class MultimodalData:
    pixel_values: torch.Tensor | None = None
    image_grid_thw: torch.Tensor | None = None
    mrope_positions: torch.Tensor | None = None
    mrope_position_delta: int | None = None

边界条件：

纯文本模型的 from_hf() 不受影响，新增字段全部有默认值。
vision_config 以原始字典形式存储，避免 frozen dataclass 嵌套问题。

子需求 2：请求协议和 tokenizer 预处理链路

场景：把"图片 + 文本"从 HTTP 请求一路传到 scheduler。

涉及文件：

文件	修改类型	影响函数/区域
`python/minisgl/server/api_server.py`	修改	`Message` 模型、`v1_completions()`
`python/minisgl/message/tokenizer.py`	修改	`TokenizeMsg`
`python/minisgl/message/backend.py`	修改	`UserMsg`
`python/minisgl/tokenizer/tokenize.py`	修改	`TokenizeManager.tokenize()`
`python/minisgl/tokenizer/server.py`	修改	tokenize 消息处理逻辑
`python/minisgl/multimodal/image_io.py`	新增	图片加载工具
`python/minisgl/multimodal/qwen3_vl_processor.py`	新增	AutoProcessor 薄封装

实现细节：

API Server 消息扩展 (api_server.py:60-62)

Message.content 从 str 改为 str | list，支持 OpenAI 风格：

class ContentPart(BaseModel):
    type: Literal["text", "image_url"]
    text: str | None = None
    image_url: dict | None = None  # {"url": "..."}

class Message(BaseModel):
    role: Literal["system", "user", "assistant"]
    content: str | list[ContentPart]

在 v1_completions 中，如果 content 包含 image_url，需要将原始消息列表传递给 tokenizer（而非只传文本）。

TokenizeMsg 扩展 (message/tokenizer.py:35-38)

@dataclass
class TokenizeMsg(BaseTokenizerMsg):
    uid: int
    text: str | List[Dict[str, str]]
    sampling_params: SamplingParams
    images: list | None = None       # PIL.Image 列表
    is_multimodal: bool = False

注意：images 字段通过 ZMQ 序列化传输，PIL.Image 需要转为 bytes 后传输。但由于 tokenizer worker 和 API server 在同一进程或可以通过 fork 共享内存，这里的 images 实际上是图片 URL/path 列表，在 tokenizer worker 侧再加载。

更好的方案：在 TokenizeMsg 中传递原始 messages（包含 image_url），在 tokenizer worker 侧统一处理图片加载和 processor 调用。

UserMsg 扩展 (message/backend.py:33-36)

@dataclass
class UserMsg(BaseBackendMsg):
    uid: int
    input_ids: torch.Tensor
    sampling_params: SamplingParams
    pixel_values: torch.Tensor | None = None
    image_grid_thw: torch.Tensor | None = None
    mrope_positions: torch.Tensor | None = None
    mrope_position_delta: int | None = None
    is_multimodal: bool = False

TokenizeManager 扩展 (tokenizer/tokenize.py)

class TokenizeManager:
    def __init__(self, tokenizer, processor=None, model_config=None):
        self.tokenizer = tokenizer
        self.processor = processor
        self.model_config = model_config

    def tokenize(self, msgs: List[TokenizeMsg]) -> List[TokenizeResult]:
        results = []
        for msg in msgs:
            if msg.is_multimodal and self.processor is not None:
                result = self._tokenize_multimodal(msg)
            else:
                result = self._tokenize_text(msg)
            results.append(result)
        return results

多模态 tokenize 路径：

解析 messages 中的 image_url，加载图片为 PIL.Image
调用 processor.apply_chat_template(messages, tokenize=False, add_generation_prompt=True) 得到文本
调用 processor(images=images, text=text, return_tensors="pt") 得到 input_ids, pixel_values, image_grid_thw
计算 MRoPE positions

tokenizer worker 改造 (tokenizer/server.py)

在创建 TokenizeManager 时，如果模型是多模态的，传入 processor。在构造 UserMsg 时，携带多模态字段。

图片加载 (multimodal/image_io.py)

def load_image(source: str) -> PIL.Image.Image:
    """支持 http(s):// URL, data:image/...;base64,..., 本地路径"""

Processor 封装 (multimodal/qwen3_vl_processor.py)

对 AutoProcessor 做薄封装，提供统一接口：

解析 OpenAI-style messages
收集图片列表
调用 processor
返回 input_ids, pixel_values, image_grid_thw

数据流：

API Server: Message(content=[{type:"text",...},{type:"image_url",...}])
  → TokenizeMsg(text=messages_list, is_multimodal=True)
  → TokenizeManager._tokenize_multimodal()
    → load_image() → PIL.Image
    → AutoProcessor() → input_ids, pixel_values, image_grid_thw
    → get_rope_index() → mrope_positions, mrope_position_delta
  → UserMsg(input_ids, pixel_values, image_grid_thw, mrope_positions, mrope_position_delta)

边界条件：

纯文本请求走原有路径，不触碰多模态逻辑。
图片 URL 无法访问时，返回明确错误。
模型不支持多模态时，收到图片请求返回 400。

子需求 3：MRoPE 与 batch 位置系统改造

场景：让 text attention 在 prefill/decode 阶段都能处理 (3, seq_len) 的 Qwen3-VL 位置编码。

涉及文件：

文件	修改类型	影响函数/区域
`python/minisgl/layers/rotary.py`	修改	`RotaryEmbedding.forward()`，新增 `MRotaryEmbedding`
`python/minisgl/layers/attention.py`	修改	`AttentionLayer.forward()`
`python/minisgl/core.py`	修改	`Req`、`Batch`
`python/minisgl/scheduler/scheduler.py`	修改	`_make_positions()`、`_process_one_msg()`
`python/minisgl/multimodal/mrope.py`	新增	`get_rope_index()`

实现细节：

MRotaryEmbedding (layers/rotary.py)

新增 MRotaryEmbedding 类，继承 RotaryEmbedding，覆盖 forward 方法：

class MRotaryEmbedding(RotaryEmbedding):
    """Rotary Embedding with Multimodal Sections (MRoPE)."""

    def __init__(self, head_size, rotary_dim, max_position_embeddings, base,
                 mrope_section, post_process=None):
        super().__init__(head_size, rotary_dim, max_position_embeddings, base, post_process)
        self.mrope_section = mrope_section

    def forward(self, positions, query, key):
        if positions.ndim == 1:
            # 文本 decode 或 text-only：沿用 flashinfer fast path
            return super().forward(positions, query, key)

        # positions shape: (3, seq_len) - MRoPE path
        assert positions.shape[0] == 3
        cos_sin = self._cos_sin_cache[positions]  # (3, seq_len, rotary_dim)
        cos, sin = cos_sin.chunk(2, dim=-1)
        # 按 mrope_section 切分并选择对应维度
        cos = torch.cat(
            [m[i] for i, m in enumerate(cos.split(self.mrope_section, dim=-1))],
            dim=-1,
        )  # (seq_len, rotary_dim/2)
        sin = torch.cat(
            [m[i] for i, m in enumerate(sin.split(self.mrope_section, dim=-1))],
            dim=-1,
        )
        # 对 q, k 应用 rotary embedding (torch native)
        q = query.view(-1, num_heads, head_dim)
        k = key.view(-1, num_kv_heads, head_dim)
        q_rot, q_pass = q[..., :rotary_dim], q[..., rotary_dim:]
        k_rot, k_pass = k[..., :rotary_dim], k[..., rotary_dim:]
        q_rot = _apply_rotary_emb(q_rot, cos, sin)
        k_rot = _apply_rotary_emb(k_rot, cos, sin)
        query = torch.cat((q_rot, q_pass), dim=-1).flatten(1)
        key = torch.cat((k_rot, k_pass), dim=-1).flatten(1)
        return query, key

_get_rope 和 get_rope 需要扩展以支持 mrope_section 参数。当 rope_scaling 中包含 mrope_section 时，返回 MRotaryEmbedding。

AttentionLayer (layers/attention.py:47)

当前 forward 方法从 ctx.batch.positions 读取 positions，传给 self.rotary.forward()。MRoPE 场景下 positions 可能是 (3, seq_len) 而不是 (seq_len,)，但 rotary 层自己判断 ndim 即可，attention 层不需大改。

Req/Batch 扩展 (core.py)

@dataclass(eq=False)
class Req:
    # ... 现有字段 ...
    pixel_values: torch.Tensor | None = None
    image_grid_thw: torch.Tensor | None = None
    mrope_positions: torch.Tensor | None = None
    mrope_position_delta: int | None = None
    is_multimodal: bool = False

Batch 新增：

@dataclass
class Batch:
    # ... 现有字段 ...
    pixel_values: torch.Tensor | None = None
    image_grid_thw: torch.Tensor | None = None
    has_multimodal: bool = False

Scheduler 位置生成 (scheduler/scheduler.py:236)

_make_positions 需要分支：

纯文本 batch：沿用原逻辑（生成顺序整数）
多模态 prefill：拼接 tokenizer 侧给出的 mrope_positions (3, seq_len)
多模态 decode：按 mrope_position_delta + cached_len 生成新 3 路位置，三路相同值

def _make_positions(batch: Batch, device: torch.device) -> torch.Tensor:
    has_mrope = any(getattr(r, 'is_multimodal', False) for r in batch.padded_reqs)
    if not has_mrope:
        # 原有逻辑
        ...
    else:
        # MRoPE 逻辑
        if batch.is_prefill:
            # 拼接各 req 的 mrope_positions
            ...
        else:
            # decode: 每个 token 的 3 路 position 相同
            ...

MRoPE 位置计算 (multimodal/mrope.py)

从 sglang get_rope_index() (rotary_embedding.py:1597) 改写一个"仅 image path"版本：

def get_rope_index(
    input_ids: torch.Tensor,          # (seq_len,)
    image_grid_thw: torch.Tensor,     # (num_images, 3)
    image_token_id: int,
    vision_start_token_id: int,
    spatial_merge_size: int,
) -> tuple[torch.Tensor, int]:
    """返回 (mrope_positions: (3, seq_len), mrope_position_delta: int)"""

核心逻辑（参考 sglang rotary_embedding.py:1645-1764）：

找到所有 vision_start_token_id 位置
文本 token 获得单调递增的 3 路相同 position
图片 token 获得 3D position：t_index, h_index, w_index
t_index = arange(llm_grid_t).expand(llm_grid_h * llm_grid_w).flatten()
h_index = arange(llm_grid_h).view(1,-1,1).expand(t,_,w).flatten()
w_index = arange(llm_grid_w).view(1,1,-1).expand(t,h,_).flatten()
mrope_position_delta = positions.max() + 1 - seq_len

边界条件：

纯文本 batch 不受影响。
decode 阶段的 positions 维度需正确处理。

子需求 4：视觉编码器与 Qwen3-VL 模型主体

场景：实现真正的 Qwen3-VL 模型结构，包括 vision encoder、deepstack 和文本主干拼接。

涉及文件：

文件	修改类型	影响函数/区域
`python/minisgl/layers/vision.py`	新增	视觉编码器组件
`python/minisgl/models/qwen3_vl.py`	新增	顶层模型
`python/minisgl/models/qwen3.py`	修改	`Qwen3Model.forward()` 支持 `inputs_embeds`

实现细节：

视觉编码器组件 (layers/vision.py)

全部使用 torch native 实现，参考 sglang qwen3_vl.py:65-474：

class Qwen3VLVisionPatchEmbed(nn.Module):
    """3D 卷积 patch embedding"""
    # Conv3d(in_channels, hidden_size, kernel=[temporal_patch_size, patch_size, patch_size])

class Qwen3VisionMLP(nn.Module):
    """标准 MLP: Linear → activation → Linear"""

class Qwen3VisionAttention(nn.Module):
    """视觉 attention，使用 F.scaled_dot_product_attention"""
    # 不使用 KV cache，不走 mini-sglang attention backend

class Qwen3VisionBlock(nn.Module):
    """Pre-norm transformer block: norm1 → attn → residual → norm2 → MLP → residual"""

class Qwen3VLVisionPatchMerger(nn.Module):
    """Patch merger: norm → Linear → GELU → Linear"""
    # spatial_merge_size=2, hidden_size = context_dim * merge_size^2

class Qwen3VLVisionModel(nn.Module):
    """完整视觉编码器"""
    # patch_embed + pos_embed + rotary_pos_emb + blocks + merger + deepstack_merger_list

Qwen3VLVisionModel 的 forward 流程（参考 sglang qwen3_vl.py:436-474）：

x = patch_embed(x) - 3D 卷积提取 patch
x += fast_pos_embed_interpolate(grid_thw) - 双线性插值绝对位置编码
rot_emb = rot_pos_emb(grid_thw) - 旋转位置编码
计算 cu_seqlens
遍历 blocks，在 deepstack_visual_indexes 处捕获中间特征
最终 merger + deepstack_mergers 输出拼接
返回 (seq_len, hidden_size * (1 + num_deepstack))

视觉 Attention 实现（不使用 mini-sglang 的 attention backend，直接用 SDPA）：

class Qwen3VisionAttention(nn.Module):
    def __init__(self, embed_dim, num_heads):
        self.qkv_proj = nn.Linear(embed_dim, embed_dim * 3, bias=True)
        self.out_proj = nn.Linear(embed_dim, embed_dim, bias=True)
        self.num_heads = num_heads
        self.head_dim = embed_dim // num_heads

    def forward(self, x, cu_seqlens, position_embeddings):
        B, S, _ = x.shape
        qkv = self.qkv_proj(x)
        q, k, v = qkv.chunk(3, dim=-1)
        # 应用旋转位置编码
        cos, sin = position_embeddings
        q = apply_rotary_emb(q, cos, sin)
        k = apply_rotary_emb(k, cos, sin)
        # reshape for multi-head attention
        q = q.view(B, S, self.num_heads, self.head_dim).transpose(1, 2)
        k = k.view(B, S, self.num_heads, self.head_dim).transpose(1, 2)
        v = v.view(B, S, self.num_heads, self.head_dim).transpose(1, 2)
        out = F.scaled_dot_product_attention(q, k, v)
        out = out.transpose(1, 2).contiguous().view(B, S, -1)
        return self.out_proj(out)

Qwen3-VL 顶层模型 (models/qwen3_vl.py)

class Qwen3VLForConditionalGeneration(BaseLLMModel):
    def __init__(self, config: ModelConfig):
        self.visual = Qwen3VLVisionModel(config.vision_config)
        self.model = Qwen3Model(config)  # 复用现有文本主干
        self.lm_head = ParallelLMHead(...)
        self.deepstack_embed_to_decoder_layer = range(
            len(config.vision_config["deepstack_visual_indexes"])
        )

    def forward(self) -> torch.Tensor:
        ctx = get_global_ctx()
        batch = ctx.batch
        input_ids = batch.input_ids

        if batch.has_multimodal and batch.is_prefill:
            # 1. 运行视觉编码器
            image_embeds = self.visual(batch.pixel_values, batch.image_grid_thw)
            # 2. 分离 input_embeds 和 deepstack_embeds
            input_embeds, deepstack_embeds = self.separate_deepstack_embeds(image_embeds)
            # 3. 获取文本 embedding
            hidden = self.model.embed_tokens(input_ids)
            # 4. 替换 image placeholder
            mask = (input_ids == self.image_token_id)
            hidden[mask] = input_embeds
            # 5. 通过 decoder layers (带 deepstack 注入)
            hidden = self._forward_with_deepstack(hidden, deepstack_embeds)
        else:
            hidden = self.model.forward(input_ids)

        return self.lm_head(hidden)

Qwen3Model 改造 (models/qwen3.py:58)

给 Qwen3Model.forward() 新增一个可选的 inputs_embeds 参数，以及 deepstack_embeds 参数：

def forward(self, input_ids: torch.Tensor,
            inputs_embeds: torch.Tensor | None = None,
            deepstack_embeds: torch.Tensor | None = None) -> torch.Tensor:
    if inputs_embeds is not None:
        x = inputs_embeds
    else:
        x = self.embed_tokens.forward(input_ids)
    residual = None
    for layer_idx, layer in enumerate(self.layers.op_list):
        x, residual = layer.forward(x, residual)
        if deepstack_embeds is not None and layer_idx in deepstack_inject_layers:
            sep = hidden_size * layer_idx
            x = x + deepstack_embeds[:, sep: sep + hidden_size]
    return self.norm.forward(x, residual)[0]

边界条件：

视觉编码器独立运行，不依赖 mini-sglang 的 attention backend 和 KV cache。
DeepStack 注入仅在前 N 个 decoder layer 发生。
不含图片的 batch 走原有 Qwen3Model.forward(input_ids) 路径。

子需求 5：权重加载与命名映射

场景：让 HF safetensors 权重能正确落到 mini-sglang 的 VL 模型结构上。

涉及文件：

文件	修改类型	影响函数/区域
`python/minisgl/models/weight.py`	修改	`load_weight()`

实现细节：

当前 weight.py:92-94 跳过 vision_tower./multi_modal_projector. 前缀的权重。对 Qwen3-VL 需要改造：

# 当前逻辑 (weight.py:92-96):
if name.startswith(("vision_tower.", "multi_modal_projector.")):
    continue
raw = f.get_tensor(name)
name = name.removeprefix("language_model.")

# 改为:
architecture = config.architectures[0] if config.architectures else ""
is_qwen3_vl = architecture == "Qwen3VLForConditionalGeneration"

if not is_qwen3_vl:
    if name.startswith(("vision_tower.", "multi_modal_projector.")):
        continue

raw = f.get_tensor(name)

if is_qwen3_vl:
    # Qwen3-VL 权重名映射
    name = name.replace("model.visual.", "visual.")
    name = name.replace("model.language_model.", "model.")
    # 视觉 QKV 合并: attn.q./attn.k./attn.v. → attn.qkv.
    # (视觉权重不走现有的 _MERGE_GROUPS，需要单独处理)
else:
    name = name.removeprefix("language_model.")

视觉编码器的权重需要特殊处理：

视觉部分的 attn.q., attn.k., attn.v. 需要合并为 attn.qkv_proj.（参考 sglang qwen3_vl.py:477-501）
视觉权重不做 TP shard（第一阶段 TP=1，但即便如此也需要确保不误触 _shard_tensor 的 split 逻辑）

具体策略：

视觉权重（visual. 前缀）：不做 TP shard，仅做 QKV merge
文本权重：沿用现有 merge + shard 逻辑
rotary_emb.inv_freq 键：跳过

边界条件：

现有 Mistral/Llama/Qwen3 文本模型的加载不受影响。
merge_buf 断言在所有权重消费后仍然成立。

子需求 6：Scheduler/Engine 运行时整合

场景：让多模态 batch 在 mini-sglang 的现有运行时中稳定跑起来。

涉及文件：

文件	修改类型	影响函数/区域
`python/minisgl/scheduler/scheduler.py`	修改	`_process_one_msg()`, `_prepare_batch()`
`python/minisgl/core.py`	修改	`Batch` (已在子需求 3 完成)
`python/minisgl/engine/engine.py`	修改	`forward_batch()`
`python/minisgl/engine/graph.py`	修改	`GraphRunner.can_use_cuda_graph()`
`python/minisgl/server/api_server.py`	修改	错误处理
`python/minisgl/tokenizer/server.py`	修改	传入 processor 和 model_config
`python/minisgl/server/args.py`	修改	可能需要新增参数

实现细节：

Scheduler (scheduler.py)

_process_one_msg() 创建 Req 时写入多模态字段：

elif isinstance(msg, UserMsg):
    # ... 现有逻辑 ...
    req = Req(...)  # 新增: pixel_values, image_grid_thw, mrope_positions, mrope_position_delta, is_multimodal

_prepare_batch() 聚合多模态字段：

def _prepare_batch(self, batch: Batch) -> ForwardInput:
    # 检查是否有多模态请求
    batch.has_multimodal = any(getattr(r, 'is_multimodal', False) for r in batch.reqs)
    if batch.has_multimodal and batch.is_prefill:
        # 聚合 pixel_values, image_grid_thw
        batch.pixel_values = torch.cat([r.pixel_values for r in batch.reqs if r.pixel_values is not None])
        batch.image_grid_thw = torch.cat([r.image_grid_thw for r in batch.reqs if r.image_grid_thw is not None])
    # ... 现有逻辑 ...

Engine (engine.py:191)

def forward_batch(self, batch: Batch, args: BatchSamplingArgs) -> ForwardOutput:
    with self.ctx.forward_batch(batch):
        if getattr(batch, 'has_multimodal', False):
            # 多模态 batch 不使用 cuda graph
            logits = self.model.forward()
        elif self.graph_runner.can_use_cuda_graph(batch):
            logits = self.graph_runner.replay(batch)
        else:
            logits = self.model.forward()
    # ...

GraphRunner (engine/graph.py)

can_use_cuda_graph 在 batch 包含多模态数据时返回 False。

API Server (api_server.py)

若模型不是 multimodal，但收到图片请求，返回 400 错误。

Tokenizer Server (tokenizer/server.py)

创建 TokenizeManager 时传入 processor 和 model_config：

processor = load_processor(tokenizer_path) if is_multimodal else None
tokenize_manager = TokenizeManager(tokenizer, processor=processor, model_config=model_config)

运行时简化（第一阶段）：

多模态请求不参与 overlap scheduling 的复杂优化
多模态请求不做 prefix cache 复用
含图片请求的 batch 统一走 eager

子需求 7：测试与回归

场景：建立最小可维护测试。

涉及文件：

文件	修改类型
`tests/core/test_qwen3_vl_config.py`	新增
`tests/core/test_qwen3_vl_mrope.py`	新增
`tests/core/test_qwen3_vl_processor.py`	新增
`tests/core/test_qwen3_vl_weight.py`	新增

测试内容：

Config 测试：ModelConfig.from_hf() 对 Qwen3-VL 配置能正确提取 image_token_id, vision_config, mrope_section 等。
MRoPE 测试：get_rope_index() 对含图片 token 的 input_ids 生成正确的 (3, seq_len) positions 和 delta。
Processor 测试：OpenAI-style content + 图片输入能转成 input_ids/pixel_values/image_grid_thw。
Weight 测试：用 fake state dict 测重命名和 merge 逻辑。

预期结果

完成后，使用以下命令启动并验证：

python -m minisgl --model-path Qwen/Qwen3-VL-2B-Instruct

通过 OpenAI-compatible API 发送：

一张本地图片 + 文本问题 → 返回正确描述
一张 URL 图片 + 文本问题 → 返回正确描述
纯文本问题 → 不回归

需求概述​

技术架构​

子需求分解​

子需求 1：配置、注册、依赖补齐​

子需求 2：请求协议和 tokenizer 预处理链路​

子需求 3：MRoPE 与 batch 位置系统改造​

子需求 4：视觉编码器与 Qwen3-VL 模型主体​

子需求 5：权重加载与命名映射​

子需求 6：Scheduler/Engine 运行时整合​

子需求 7：测试与回归​

预期结果​

需求概述

技术架构

子需求分解

子需求 1：配置、注册、依赖补齐

子需求 2：请求协议和 tokenizer 预处理链路

子需求 3：MRoPE 与 batch 位置系统改造

子需求 4：视觉编码器与 Qwen3-VL 模型主体

子需求 5：权重加载与命名映射

子需求 6：Scheduler/Engine 运行时整合

子需求 7：测试与回归

预期结果